package net.ustc.javamemcached.pipeline;

import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import net.ustc.javamemcached.Key;
import net.ustc.javamemcached.memorystore.MemoryEntry;
import net.ustc.javamemcached.memorystore.IMemoryManager;
import net.ustc.javamemcached.processor.IProtocolProcessor;
import net.ustc.javamemcached.protocol.ProtocolMessage;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MemcachedProtocolMessageHandler extends
		SimpleChannelUpstreamHandler {
	final Logger logger = LoggerFactory
			.getLogger(MemcachedProtocolMessageHandler.class);

	public final AtomicInteger curr_conns = new AtomicInteger();
	public final AtomicInteger total_conns = new AtomicInteger();

	/**
	 * The following state variables are universal for the entire daemon. These
	 * are used for statistics gathering. In order for these values to work
	 * properly, the handler _must_ be declared with a ChannelPipelineCoverage
	 * of "all".
	 */
	public final String version;

	public final int idle_limit;

	/**
	 * The actual physical data storage.
	 */
	private final IMemoryManager memoryManager;
	public final Map<ChannelBuffer, IProtocolProcessor> processors;

	/**
	 * The channel group for the entire daemon, used for handling global cleanup
	 * on shutdown.
	 */
	private final DefaultChannelGroup channelGroup;

	
	
	public MemcachedProtocolMessageHandler(String version, int idleLimit, IMemoryManager memoryManager,
			Map<ChannelBuffer, IProtocolProcessor> processors,
			DefaultChannelGroup channelGroup) {
		super();
		this.version = version;
		idle_limit = idleLimit;
		this.memoryManager = memoryManager;
		this.processors = processors;
		this.channelGroup = channelGroup;
	}

	/**
	 * On open we manage some statistics, and add this connection to the channel
	 * group.
	 * 
	 * @param channelHandlerContext
	 * @param channelStateEvent
	 * @throws Exception
	 */
	@Override
	public void channelOpen(ChannelHandlerContext channelHandlerContext,
			ChannelStateEvent channelStateEvent) throws Exception {
		total_conns.incrementAndGet();
		curr_conns.incrementAndGet();
		channelGroup.add(channelHandlerContext.getChannel());
	}

	/**
	 * On close we manage some statistics, and remove this connection from the
	 * channel group.
	 * 
	 * @param channelHandlerContext
	 * @param channelStateEvent
	 * @throws Exception
	 */
	@Override
	public void channelClosed(ChannelHandlerContext channelHandlerContext,
			ChannelStateEvent channelStateEvent) throws Exception {
		curr_conns.decrementAndGet();
		channelGroup.remove(channelHandlerContext.getChannel());
	}

	/**
	 * The actual meat of the matter. Turn CommandMessages into executions
	 * against the physical cache, and then pass on the downstream messages.
	 * 
	 * @param channelHandlerContext
	 * @param messageEvent
	 * @throws Exception
	 */

	@Override
	@SuppressWarnings("unchecked")
	public void messageReceived(ChannelHandlerContext channelHandlerContext,
			MessageEvent messageEvent) throws Exception {
	}

	/**
	 * @return the current time in seconds (from epoch), used for expiries, etc.
	 */
	private static int Now() {
		return (int) (System.currentTimeMillis() / 1000);
	}

}
